package com.hanserwei.community.config;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;

@Configuration
public class RedisConfig {

    private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
        Kryo kryo = new Kryo();
        kryo.setRegistrationRequired(false);
        // 如果需要，可以在这里注册特定的类
        return kryo;
    });

    /**
     * 定义 Kryo 序列化器 Bean
     * 这样就可以在 RedisTemplate 中注入和使用了
     * <p>如果你只是打算存验证码，那么完全不需要Kryo。但是这是通用模版，我爱用！</p>
     */
    @Bean
    public RedisSerializer<Object> kryoRedisSerializer() {
        return new RedisSerializer<>() {
            @Override
            public byte[] serialize(Object o) throws SerializationException {
                if (o == null) {
                    return new byte[0];
                }
                Kryo kryo = kryoThreadLocal.get();
                try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
                     Output output = new Output(byteArrayOutputStream)) {
                    kryo.writeObject(output, o);
                    output.flush();
                    return byteArrayOutputStream.toByteArray();
                } catch (Exception e) {
                    throw new SerializationException("Kryo serialize error", e);
                }
            }

            @Override
            public Object deserialize(byte[] bytes) throws SerializationException {
                if (bytes == null || bytes.length == 0) {
                    return null;
                }
                Kryo kryo = kryoThreadLocal.get();
                try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                     Input input = new Input(bais)) {
                    return kryo.readClassAndObject(input);
                } catch (Exception e) {
                    throw new SerializationException("Kryo deserialize error", e);
                }
            }
        };
    }

    /**
     * 修改后的 RedisTemplate 配置
     * Key 仍使用 String 序列化，Value 改用 Kryo 序列化
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(
            RedisConnectionFactory factory,
            RedisSerializer<Object> kryoRedisSerializer) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);

        StringRedisSerializer stringSerializer = new StringRedisSerializer();

        // 设置 Key 的序列化器
        template.setKeySerializer(stringSerializer);
        template.setHashKeySerializer(stringSerializer);

        // 设置 Value 的序列化器为 Kryo
        template.setValueSerializer(kryoRedisSerializer);
        template.setHashValueSerializer(kryoRedisSerializer);

        template.afterPropertiesSet();
        return template;
    }
}